package cn.com.dcsgo.service.impl;

import cn.com.dcsgo.confit.EsImportConfig;
import cn.com.dcsgo.constant.EsConstant;
import cn.com.dcsgo.domain.Prod;
import cn.com.dcsgo.domain.ProdTagReference;
import cn.com.dcsgo.mapper.ProdCommMapper;
import cn.com.dcsgo.mapper.ProdMapper;
import cn.com.dcsgo.mapper.ProdTagReferenceMapper;
import cn.com.dcsgo.model.CommStatistics;
import cn.com.dcsgo.model.ProdEs;
import cn.com.dcsgo.service.ImportService;
import cn.com.dcsgo.utils.ThreadPoolUtils;
import cn.hutool.core.bean.BeanUtil;
import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch.core.BulkRequest;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.elasticsearch.indices.CreateIndexResponse;
import co.elastic.clients.elasticsearch.indices.ElasticsearchIndicesClient;
import co.elastic.clients.elasticsearch.indices.ExistsRequest;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.CommandLineRunner;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.scheduling.annotation.Scheduled;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.springframework.util.ObjectUtils;

import java.io.IOException;
import java.math.BigDecimal;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.ArrayList;
import java.util.Date;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.stream.Collectors;

/**
 * 向ES中导入商品信息
 *
 * @author Dcsgo
 * @since 2025/10/25 10:39
 */
@Slf4j
@Service
public class ImportServiceImpl implements ImportService, CommandLineRunner {
    @Resource
    private EsImportConfig esImportConfig;

    @Resource
    private ElasticsearchClient elasticsearchClient;

    @Resource
    private StringRedisTemplate stringRedisTemplate;


    @Resource
    private ProdMapper prodMapper;

    @Resource
    private ProdTagReferenceMapper prodTagReferenceMapper;

    @Resource
    private ProdCommMapper prodCommMapper;

    /**
     * 上次导入时间（用于增量更新->在此时间之后的数据才进行导入）
     */
    private Date importDate;
    private final SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
    private final int pageSize = 100;

    @Override
    public void importAll() throws IOException, InterruptedException {
        // 没有开启批量导入，不进行导入操作
        if (!esImportConfig.getFlag()) {
            log.info("Dcsgo:ES全量导入在配置文件中已关闭");
            return;
        }
        //0.记录导入时间
        importDate = new Date();
        // 后续当前服务可能有多个服务器运行，使用Redis存储导入时间以同步该时间
        stringRedisTemplate.opsForValue().set(EsConstant.UPDATE_IMPORT_TIME_KEY, sdf.format(importDate));
        log.info("Dcsgo:ES全量导入开始");
        //1.创建商品数据索引
        createIndex();
        //2.分页导入商品数据（导入数据量大时，深分页如何解决？子查询查分页数据ID，再内连接ID查询）
        //查询需要导入的商品数据总条数
        Long totalCount = getTotalCount(null);
        if (totalCount.equals(0L)) {
            log.info("Dcsgo:ES全量导入时没有需要导入的数据");
            return;
        }
        //计算总页数
        int totalPage = (int) Math.ceil(totalCount * 1.0 / pageSize);
        //关闭副本和刷新
        closeReplicasAndRefresh();
        //线程池处理分页导入
        CountDownLatch countDownLatch = new CountDownLatch(totalPage);
        for (int i = 0; i < totalPage; i++) {
            int index = i;
            ThreadPoolUtils.poolExecutor.submit(() -> {
                try {
                    fetchImportProdEs(index + 1, pageSize, null);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
        } catch (InterruptedException e) {
            log.error("Dcsgo:ES全量导入-线程池处理分页导入失败，失败原因：{}", e.getMessage());
            throw e;
        }
        log.info("Dcsgo:ES全量导入成功，共导入了{}条数据，共{}页", totalCount, totalPage);
        //打开副本和刷新
        openReplicasAndRefresh();
    }

    /**
     * 分页导入商品数据到ES
     *
     * @param pageNumber 页码
     * @param pageSize   分页大小
     * @param updateTime 更新时间（不为null时只导入晚于该时间的商品数据）
     */
    private void fetchImportProdEs(int pageNumber, int pageSize, Date updateTime) throws IOException {
        //分页查询商品数据
        List<Prod> prodList = prodMapper.getByPage((pageNumber - 1) * pageSize, pageSize, updateTime);
        if (CollectionUtils.isEmpty(prodList)) {
            return;
        }
        List<Long> prodIds = prodList.stream().map(Prod::getProdId).toList();
        //获取商品活动ID列表
        List<ProdTagReference> prodTagReferences = prodTagReferenceMapper.selectList(
                new LambdaQueryWrapper<ProdTagReference>()
                        .select(ProdTagReference::getTagId, ProdTagReference::getProdId)
                        .eq(ProdTagReference::getStatus, 1)
                        .in(ProdTagReference::getProdId, prodIds)
        );
        //把List<ProdTagReference> 转换成 Map<ProdId,List<TagId>>
        Map<Long, List<Long>> prodTagMap = prodTagReferences.stream().collect(
                Collectors.groupingBy(ProdTagReference::getProdId,
                        Collectors.mapping(ProdTagReference::getTagId, Collectors.toList())
                )
        );
        //获取商品总评和好评数
        List<CommStatistics> commStatisticsList = prodCommMapper.selectStatisticsByProdIds(prodIds);
        //把List<CommStatistics> 转化成 Map<prodId,CommStatistics>
        Map<Long, CommStatistics> prodCommStatisticsMap = commStatisticsList.stream().collect(
                Collectors.toMap(CommStatistics::getProdId, commStatistics -> commStatistics)
        );
        //把List<Prod>转换成List<ProdEs>
        List<ProdEs> prodEsList = prodList.stream().map(prod -> {
            ProdEs prodEs = new ProdEs();
            BeanUtil.copyProperties(prod, prodEs);
            //设置活动ID列表
            List<Long> tagIds = prodTagMap.get(prod.getProdId());
            if (!CollectionUtils.isEmpty(tagIds)) {
                prodEs.setTagList(tagIds);
            }
            //处理评论数据
            CommStatistics commStatistics = prodCommStatisticsMap.get(prod.getProdId());
            if (!ObjectUtils.isEmpty(commStatistics)) {
                prodEs.setPraiseNumber((long) commStatistics.getGoodCount());
                //计算好评率（10%->对应的计算结果就是10）
                if (commStatistics.getGoodCount() != 0) {
                    double positiveRating = Math.ceil(commStatistics.getGoodCount() * 1.0 / commStatistics.getAllCount() * 100);
                    prodEs.setPositiveRating(new BigDecimal(positiveRating));
                }
            }
            return prodEs;
        }).toList();
        //把数据导入到ES中
        List<BulkOperation> bulkOperations = new ArrayList<>(prodEsList.size());
        for (ProdEs prodEs : prodEsList) {
            bulkOperations.add(BulkOperation.of(doc -> doc.index(
                    c -> c.id(prodEs.getProdId().toString()).document(prodEs)
            )));
        }
        BulkRequest request = BulkRequest.of(bulk -> bulk.index(EsConstant.PROD_ES_INDEX).operations(bulkOperations));
        try {
            elasticsearchClient.bulk(request);
            log.info("Dcsgo:ES分页导入数据成功，第{}页，共{}条数据", pageNumber, prodEsList.size());
        } catch (IOException e) {
            log.error("Dcsgo:ES分页导入数据失败，第{}数据导入失败，失败原因：{}", pageNumber, e.getMessage());
            throw e;
        }
    }

    /**
     * 打开ES副本和刷新
     */
    private void openReplicasAndRefresh() throws IOException {
        if (!checkIndexExists(EsConstant.PROD_ES_INDEX)) {
            return;
        }
        try {
            ElasticsearchIndicesClient indices = elasticsearchClient.indices();
            indices.putSettings(settings -> settings
                    .index(EsConstant.PROD_ES_INDEX)
                    .settings(setting -> setting
                            .numberOfReplicas("2")
                            .refreshInterval(value -> value.time("1s"))
                    )
            );
            log.info("Dcsgo:打开ES索引[{}]的副本和刷新成功", EsConstant.PROD_ES_INDEX);
        } catch (IOException e) {
            log.error("Dcsgo:打开ES索引[{}]的副本和刷新失败，失败原因：{}", EsConstant.PROD_ES_INDEX, e.getMessage());
            throw e;
        }
    }

    /**
     * 关闭ES副本和刷新
     */
    private void closeReplicasAndRefresh() throws IOException {
        if (!checkIndexExists(EsConstant.PROD_ES_INDEX)) {
            return;
        }
        try {
            ElasticsearchIndicesClient indices = elasticsearchClient.indices();
            indices.putSettings(settings -> settings
                    .index(EsConstant.PROD_ES_INDEX)
                    .settings(setting -> setting
                            .numberOfReplicas("0")
                            .refreshInterval(value -> value.time("-1"))
                    )
            );
            log.info("Dcsgo:关闭ES索引[{}]的副本和刷新成功", EsConstant.PROD_ES_INDEX);
        } catch (IOException e) {
            log.error("Dcsgo:关闭ES索引[{}]的副本和刷新失败，失败原因：{}", EsConstant.PROD_ES_INDEX, e.getMessage());
            throw e;
        }
    }

    /**
     * 查询商品数据总条数（状态为正常的）
     *
     * @param updateTime 商品表中数据的更新时间
     * @return 如果 updateTime 不为 null 则返回更新时间晚于 updateTime 的记录数，反之返回所有记录数
     */
    private Long getTotalCount(Date updateTime) {
        return prodMapper.selectCount(
                new LambdaQueryWrapper<Prod>()
                        .eq(Prod::getStatus, 1)
                        .ge(updateTime != null, Prod::getUpdateTime, updateTime)
        );
    }

    /**
     * 创建商品数据索引，已存在则不创建
     */
    private void createIndex() throws IOException {
        if (checkIndexExists(EsConstant.PROD_ES_INDEX)) {
            return;
        }
        CreateIndexRequest request = CreateIndexRequest.of(
                index -> index.index(EsConstant.PROD_ES_INDEX)
                        .mappings(mapping -> mapping
                                .properties("prodId", value -> value.long_(l -> l))
                                .properties("prodName", title -> title.text(t -> t.analyzer("ik_max_word").searchAnalyzer("ik_smart")))
                                .properties("shopId", value -> value.long_(l -> l))
                                .properties("price", value -> value.double_(d -> d))
                                .properties("brief", brief -> brief.text(t -> t.analyzer("ik_max_word").searchAnalyzer("ik_smart")))
                                .properties("pic", pic -> pic.text(t -> t))
                                .properties("categoryId", value -> value.long_(l -> l))
                                .properties("soldNum", value -> value.integer(i -> i))
                                .properties("tagList", value -> value.long_(l -> l))
                                .properties("positiveRating", value -> value.double_(d -> d))
                                .properties("praiseNumber", value -> value.long_(l -> l))
                        )
        );
        CreateIndexResponse response;
        try {
            response = elasticsearchClient.indices().create(request);
            log.info("Dcsgo:ES索引[{}]创建状态:{}", EsConstant.PROD_ES_INDEX, response.toString());
        } catch (IOException e) {
            log.error("Dcsgo:ES索引[{}]创建失败，异常原因：{}", EsConstant.PROD_ES_INDEX, e.getMessage());
            throw e;
        }
    }

    /**
     * 检查ES索引是否存在
     *
     * @param indexName ES 索引名
     * @return 存在返回true否则返回false
     */
    private boolean checkIndexExists(String indexName) throws IOException {
        ExistsRequest existsRequest = ExistsRequest.of(ext -> ext.index(indexName));
        try {
            BooleanResponse existsResponse = elasticsearchClient.indices().exists(existsRequest);
            return existsResponse.value();
        } catch (IOException e) {
            log.error("Dcsgo:判断ES索引[{}]是否存在失败，失败原因：{}", indexName, e.getMessage());
            throw e;
        }
    }

    @Override
    @Scheduled(fixedDelay = EsConstant.ONE_HOUR)
    public void importUpdate() throws ParseException, InterruptedException {
        //0.获取上次导入时间
        String lastImportDateStr = stringRedisTemplate.opsForValue().get(EsConstant.UPDATE_IMPORT_TIME_KEY);
        importDate = sdf.parse(lastImportDateStr);
        log.info("Dcsgo:ES增量导入开始");
        //查询需要导入的商品数据总条数
        Long totalCount = getTotalCount(importDate);
        //计算总页数
        int totalPage = (int) Math.ceil(totalCount * 1.0 / pageSize);
        //线程池处理分页导入
        CountDownLatch countDownLatch = new CountDownLatch(totalPage);
        for (int i = 0; i < totalPage; i++) {
            int index = i;
            ThreadPoolUtils.poolExecutor.submit(() -> {
                try {
                    fetchImportProdEs(index + 1, pageSize, importDate);
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
                countDownLatch.countDown();
            });
        }
        try {
            countDownLatch.await();
            //记录导入时间
            importDate = new Date();
            stringRedisTemplate.opsForValue().set(EsConstant.UPDATE_IMPORT_TIME_KEY, sdf.format(importDate));
        } catch (InterruptedException e) {
            log.error("Dcsgo:ES增量导入-线程池处理分页导入失败，失败原因：{}", e.getMessage());
            throw e;
        }
        log.info("Dcsgo:ES增量导入成功，共导入了{}条数据，共{}页", totalCount, totalPage);
    }

    @Override
    public void run(String... args) throws Exception {
        importAll();
    }
}
